package com.zho.common.consumer.queue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

import com.zho.common.consumer.DefaultInDataService;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

public abstract class DefaultQueueInDataService<T> extends DefaultInDataService<T> implements IQueueInDataService<T>, InitializingBean, DisposableBean {
    private final int MAX_RECEIVE_ITEMS = 20;

    protected Queue<T> QUEUE_DATA = new LinkedList<T>();

    private int maxReceiveItem = MAX_RECEIVE_ITEMS;

    /*
     * (non-Javadoc)
     * 
     * @see com.ipa.fss.gwcommons.consumer.IInDataService#checkAvailable()
     */
    public boolean checkAvailable() {
        return !QUEUE_DATA.isEmpty();
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.ipa.fss.gwcommons.consumer.IInDataService#recieve()
     */
    public List<T> recieve() {
        int count = 0;
        List<T> listData = new ArrayList<T>();
        T msg = null;
        while (count < maxReceiveItem) {
            msg = QUEUE_DATA.poll();
            if (msg != null) {
                listData.add(msg);
            } else {
                break;
            }
            count++;
        }
        return listData;
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
     */
    public void afterPropertiesSet() throws Exception {
        maxReceiveItem = (maxReceiveItem < 5 ? MAX_RECEIVE_ITEMS : maxReceiveItem);
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.springframework.beans.factory.DisposableBean#destroy()
     */
    public void destroy() throws Exception {
        // TODO: can overwrite on sub-classes
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.ipa.fss.gwcommons.consumer.queue.IQueueInDataService#add(java.lang.Object)
     */
    public void add(T data) {
        if (data != null) {
            QUEUE_DATA.add(data);

            // +++ send notify to Consumer
            this.sendNotify();
            // ---
        }
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.ipa.fss.gwcommons.consumer.queue.IQueueInDataService#add(java.util.Collection)
     */
    public void add(Collection<T> data) {
        if (data != null && data.size() > 0) {
            QUEUE_DATA.addAll(data);
            // +++ send notify to Consumer
            this.sendNotify();
            // ---
        }
    }

    /**
     * @return the maxReceiveItem
     */
    public int getMaxReceiveItem() {
        return maxReceiveItem;
    }

    /**
     * @param maxReceiveItem
     *            the maxReceiveItem to set
     */
    public void setMaxReceiveItem(int maxReceiveItem) {
        this.maxReceiveItem = maxReceiveItem;
    }
}
